package external_sort_algorithms;

import java.io.BufferedWriter;
import java.io.IOException;
import java.util.Comparator;
import java.util.Iterator;
import java.util.PriorityQueue;
import java.util.List;

import utils.ArgumentAssertion;

public class ScalableSortUtils {

	public static <E> void mergeAndWrite(List<Iterator<E>> iterators,
	        Comparator<E> comparator, BufferedWriter writer) throws IOException {
		
		ArgumentAssertion.assertNotEmpty("iterators", iterators);
		ArgumentAssertion.assertNotNull("comparator", comparator);
		ArgumentAssertion.assertNotNull("writer", writer);

		PriorityQueue<ScalableSortItem<E>> queue = new PriorityQueue<ScalableSortItem<E>>(
		        iterators.size(), new ScalableSortItemComparator<E>(comparator));

		for (Iterator<E> iterator : iterators) {
			if (iterator.hasNext()) {
				E element = iterator.next();
				ScalableSortItem<E> item = new ScalableSortItem<E>(element,
				        iterator);
				queue.add(item);
			}
		}

		while (!(queue.isEmpty())) {
			ScalableSortItem<E> item = queue.poll();

			E element = item.getElement();
			Iterator<E> iterator = item.getIterator();

			writer.write(element.toString());
			writer.newLine();

			if (iterator.hasNext()) {
				element = iterator.next();
				item = new ScalableSortItem<E>(element, iterator);
				queue.add(item);
			}
		}

		writer.flush();
	}

	private static class ScalableSortItem<E> {

		private E element;
		private Iterator<E> iterator;

		public ScalableSortItem(E element, Iterator<E> iterator) {
			this.element = element;
			this.iterator = iterator;
		}

		public E getElement() {
			return element;
		}

		public Iterator<E> getIterator() {
			return iterator;
		}
	}

	private static class ScalableSortItemComparator<E> implements
	        Comparator<ScalableSortItem<E>> {

		private Comparator<E> comparator;

		public ScalableSortItemComparator(Comparator<E> comparator) {
			this.comparator = comparator;
		}

		@Override
		public int compare(ScalableSortItem<E> o1, ScalableSortItem<E> o2) {
			return this.comparator.compare(o1.element, o2.element);
		}
	}
}
